/*
 * @Author: rock
 * @Date: 2025-05-23 23:12:35
 * @LastEditors: rock 
 * @LastEditTime: 2025-05-25 16:08:08
 * @FilePath: /002ThreadPool/src/threadpool.cpp
 * @Description: Ctrl+Num1-> For Document; Ctrl+Num2-> For Function
 *
 * Copyright (c) 2025 by rock, All Rights Reserved.
 */
#include "threadpool.h"

/***************************************************ThreadPool方法实现*********************************************************************** */
size_t ThreadPool::count_tasks_ = 0;        // 初始化为0, 任务计数
size_t ThreadPool::count_submit_tasks_ = 0; // 提交任务计数
const int TASK_THRESHHOLD = 4;           // 最大任务数量
const int TASK_CONDUCTER_WAIT_TIME = 1;     // 生产者等待时间, 过了这个时间线程池返回给生产者, 不能再等了, 任务队列还是满的
const int TASK_CONSUMER__WAIT_TIME = 10;    // 消费者不要再等了, 已经有5秒队列是空的了, 队列已经5秒没有生产者给任务了, 可以退出了
// 构造
ThreadPool::ThreadPool()
    : initThreadSize_(0) // CPU核心数量
      ,
      taskSize_(0) // 开始的时候, 任务的数量是0
      ,
      taskQueMaxThreshHold_(TASK_THRESHHOLD) // 任务队列中的最大任务数量是1024个(TASK_THRESHHOLD)
      ,
      poolMode_(PoolMode::MOD_FIXED) // 固定数量模式的线程池
{
    LOG("ThreadPool() 初始化线程池");
}

// 析构
ThreadPool::~ThreadPool()
{
    LOG("线程池的析构");
}

// 设置线程池工作的增长模式
void ThreadPool::setMode(PoolMode mode)
{
    poolMode_ = mode;
}

// 设置task任务队列上限阈值
void ThreadPool::setTaskQueMaxThreshHold(int threshhold)
{
    taskQueMaxThreshHold_ = threshhold;
}

// 给线程池提交任务
// 用户给任务队列传入任务, 生产任务
// std::shared_ptr<Task> sp; 要求用户传入智能指针类型
Result ThreadPool::submitTask(std::shared_ptr<Task> sp)
{
    // 获取锁, 任务队列并不是线程安全的
    std::unique_lock<std::mutex> lock(taskQueMtx_);
    // // 线程的通信, 通过信号量等待任务队列有空任务, 然后把新任务交给任务队列
    // while (taskQue_.size() == taskQueMaxThreshHold_)
    // {
    //     //任务队列中的任务数量, 等于了最大任务数量, 也就是说不能再向任务队列中继续存放任务了
    //     notFull_.wait(lock);
    // }
    // //添加条件, 用户提交任务阻塞时间最长不能超过1秒, 否则任务提交失败, 说明任务队列现在是满的, 返回, 不能让用户一直阻塞
    // notFull_.wait(lock, [&]()->bool{
    //    return taskQue_.size() < taskQueMaxThreshHold_; //任务队列中的任务数量, 等于了最大任务数量, 也就是说不能再向任务队列中继续存放任务了
    // });
    // 如果有空余, 把任务放入任务队列中

    // //添加条件, 用户提交任务阻塞时间最长不能超过1秒, 否则任务提交失败, 说明任务队列现在是满的, 返回, 不能让用户一直阻塞
    bool ret_wait_for = notFull_.wait_for(
        lock,
        std::chrono::seconds(TASK_CONDUCTER_WAIT_TIME), // 等待1秒
        [&]() -> bool
        {
            LOG("等待的时候, 队列中含有的任务数量是: taskQue_.size() = " << taskQue_.size() << ", taskQueMaxThreshHold_ = " << taskQueMaxThreshHold_);
            return taskQue_.size() < taskQueMaxThreshHold_; // 任务队列中的任务数量, 等于了最大任务数量, 也就是说不能再向任务队列中继续存放任务了
        });

    if (!ret_wait_for)
    {
        // 等待1秒钟, wait_for条件依然没有满足, 任务队列仍然没有空闲, 提交任务失败
        LOG("task queue is full, submit task failed --------------------------------");
        return Result(sp, false);
    }
    count_submit_tasks_++;
    LOG("任务提交成功, 提交了" << count_submit_tasks_ << "个任务");

    taskQue_.emplace(sp);
    taskSize_++; // 任务数量加1

    // 因为新放了任务, 任务队列肯定不空了, 那么提醒消费者来消费任务notEmpty.notify
    notEmpty_.notify_all(); // 通知

    //根据任务数量和空闲线程的数量, 判断是否需要创建新的线程出来

    return Result(sp);
}

// 开启线程池
// 用户不传递参数有默认参数, 参数的数量是CPU核心数
// 用户穿了就是用户自定义数量
void ThreadPool::start(int initThreadSize)
{
    // 记录线程池初始线程的个数
    initThreadSize_ = initThreadSize;

    // 创建线程对象std::vector<Thread *> threads_;
    for (int i = 0; i < initThreadSize_; i++)
    {
        // 创建thread线程对象的时候, 把线程函数给到thread线程对象
        auto ptr = new Thread(std::bind(&ThreadPool::threadFunc, this, i + 1), i + 1);
        threads_.emplace_back(std::move(ptr)); // 资源转移
    }

    // 启动所有线程
    // 线程对象std::vector<Thread *> threads_;
    for (int i = 0; i < initThreadSize_; i++)
    {
        // 需要执行一个线程函数
        threads_[i]->start();
    }
}

// 定义线程函数
// 线程池中的所有线程, 从任务队列中消费任务
void ThreadPool::threadFunc(int id)
{
    // 线程池中的每一个线程都有自己的线程id
    LOG("begin, 第" << id << "个, tId:" << std::this_thread::get_id());

    // 线程池中的线程一直在取任务, 一直在做事情
    for (;;)
    {
        std::shared_ptr<Task> task = nullptr;
        {
            // 先获取锁
            std::unique_lock<std::mutex> lock(taskQueMtx_);

            // 等待notEmpty
            LOG("线程开始等待任务, 尝试获取任务");

            bool ret_wait_for = notEmpty_.wait_for(lock,
                                                   std::chrono::seconds(TASK_CONSUMER__WAIT_TIME), // 等待5秒没有任务就退出
                                                   [&]() -> bool
                                                   {
                                                       return taskQue_.size() > 0;
                                                   });

            if (!ret_wait_for)
            {
                LOG("等待了10s, 还是没有任务, 线程退出");
                break;
            }

            // notEmpty_.wait(lock,

            //                [&]() -> bool
            //                {
            //                    return taskQue_.size() > 0;
            //                });

            LOG("线程等待任务结束, 获取任务成功");

            // 从任务队列中取一个任务
            task = taskQue_.front();
            taskQue_.pop();
            taskSize_--;

            // 当前线程负责执行这个任务, 取走任务后就直接释放锁就可以了, 执行任务就不需要这把锁了
            // 出作用域就释放了锁
            // 通知
            // 如果依然有剩余任务, 继续通知其他线程执行任务
            if (taskQue_.size() > 0)
            {
                // 通知其他线程, 任务队列依然还有任务, 同志们继续取任务, 做任务
                notEmpty_.notify_all();
            }
            // 通知生产者队列不满, 可以放任务了
            notFull_.notify_all();
        }

        if (task != nullptr)
        {
            count_tasks_++;
            LOG("run..., 第 " << count_tasks_ << "个任务, 执行");
            //task->run();
            task->exec();
        }
    }

    LOG("  end, 第" << id << "个线程, tId:" << std::this_thread::get_id());
}

/***************************************************Thread方法实现*************************************************************/

// 启动线程
void Thread::start()
{
    // 执行一个线程函数
    LOG("线程池中的线程start()方法实现, 这里做了线程分离");
    // 创建一个线程, 执行一个线程函数
    // 把线程池传递过来的函数对象执行一下就可以了
    std::thread t(func_, id_); // 因为线程函数和线程对象有绑定关系, 所以线程对象要和主线程分离
    t.detach();                // 分离线程, linux下的pthread_thread
    // t.join();
}

// Thread构造
Thread::Thread(ThreadFunc func, int id)
    : func_(func), id_(id)

{
    LOG("Thread::Thread(ThreadFunc func)构造, id = " << id_);
}
// Thread析构
Thread::~Thread()
{
    LOG("Thread::~Thread()析构, id = " << id_);
}


/***************************************************Result方法实现*************************************************************/
Result::Result(std::shared_ptr<Task> task, bool isValid)
            :task_(task)
            ,isValid_(isValid)
{
    task_->setResult(this);
}
// 移动构造
Result::Result(Result&& other) noexcept 
    : any_(std::move(other.any_))
    // , isValid_(other.isValid_) 
{
    isValid_.store(other.isValid_.load());
}



/***************************************************Task方法实现*************************************************************/
void Task::setResult(Result* res)
{
    result_ = res;
}


void Task::exec()
{
    if(result_!=nullptr)
    {
        result_->setVal(run());//多态调用, run()
    }    
}

Task::Task()
    :result_(nullptr)
{

}